package com.bestvike.stone.core.concurrent;

import com.bestvike.linq.exception.NotSupportedException;
import com.bestvike.stone.core.concurrent.jdk.RejectedExecutionHandler;
import com.bestvike.stone.core.concurrent.jdk.ThreadPoolExecutor;
import com.bestvike.stone.core.reflect.ReflectionUtils;
import org.springframework.util.Assert;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * Created by 许崇雷 on 2018-08-28.
 */
public final class ThreadPoolTaskScheduler extends org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler {
    private final Object monitor = new Object();
    private ScheduledThreadPoolExecutorEx threadPoolExecutor;
    private boolean removeOnCancelPolicy = false;
    private int corePoolSize = 1;
    private int maxPoolSize = Integer.MAX_VALUE;
    private int keepAliveSeconds = 60;
    private int queueCapacity = Integer.MAX_VALUE;
    private boolean allowCoreThreadTimeOut = false;
    private RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();


    //region init

    /**
     * Create a new {@link ScheduledExecutorService} instance.
     * <p>The default implementation creates a {@link ScheduledThreadPoolExecutor}.
     * Can be overridden in subclasses to provide custom {@link ScheduledExecutorService} instances.
     *
     * @param poolSize                 the specified pool size
     * @param threadFactory            the ThreadFactory to use
     * @param rejectedExecutionHandler the RejectedExecutionHandler to use
     * @return a new ScheduledExecutorService instance
     * @see #afterPropertiesSet()
     * @see java.util.concurrent.ScheduledThreadPoolExecutor
     */
    @Override
    protected ScheduledThreadPoolExecutorEx createExecutor(int poolSize, ThreadFactory threadFactory, java.util.concurrent.RejectedExecutionHandler rejectedExecutionHandler) {
        return new ScheduledThreadPoolExecutorEx(poolSize, threadFactory, this.rejectedExecutionHandler);
    }

    /**
     * Create the target {@link java.util.concurrent.ExecutorService} instance.
     * Called by {@code afterPropertiesSet}.
     *
     * @param threadFactory            the ThreadFactory to use
     * @param rejectedExecutionHandler the RejectedExecutionHandler to use
     * @return a new ExecutorService instance
     * @see #afterPropertiesSet()
     */
    @Override
    protected ExecutorService initializeExecutor(ThreadFactory threadFactory, java.util.concurrent.RejectedExecutionHandler rejectedExecutionHandler) {
        this.threadPoolExecutor = this.createExecutor(this.corePoolSize, threadFactory, rejectedExecutionHandler);
        ReflectionUtils.setFieldValue(this, "scheduledExecutor", this.threadPoolExecutor);
        if (this.removeOnCancelPolicy)
            this.threadPoolExecutor.setRemoveOnCancelPolicy(true);
        this.threadPoolExecutor.setMaximumPoolSize(this.maxPoolSize);
        this.threadPoolExecutor.setKeepAliveTime(this.keepAliveSeconds, TimeUnit.SECONDS);
        this.threadPoolExecutor.setQueueCapacity(this.queueCapacity);
        this.threadPoolExecutor.allowCoreThreadTimeOut(this.allowCoreThreadTimeOut);
        return this.threadPoolExecutor;
    }

    //endregion


    //region config

    /**
     * Return the current setting for the remove-on-cancel mode.
     * <p>Requires an underlying {@link java.util.concurrent.ScheduledThreadPoolExecutor}.
     */
    @Override
    public boolean isRemoveOnCancelPolicy() {
        synchronized (this.monitor) {
            return this.threadPoolExecutor == null
                    ? this.removeOnCancelPolicy
                    : this.getScheduledThreadPoolExecutorEx().getRemoveOnCancelPolicy();
        }
    }

    /**
     * Set the remove-on-cancel mode on {@link java.util.concurrent.ScheduledThreadPoolExecutor} (JDK 7+).
     * <p>Default is {@code false}. If set to {@code true}, the target executor will be
     * switched into remove-on-cancel mode (if possible, with a soft fallback otherwise).
     * <p><b>This setting can be modified at runtime, for example through JMX.</b>
     */
    @Override
    public void setRemoveOnCancelPolicy(boolean removeOnCancelPolicy) {
        synchronized (this.monitor) {
            this.removeOnCancelPolicy = removeOnCancelPolicy;
            if (this.threadPoolExecutor != null)
                this.threadPoolExecutor.setRemoveOnCancelPolicy(removeOnCancelPolicy);
        }
    }

    /**
     * Return the ThreadPoolExecutor's core pool size.
     */
    public int getCorePoolSize() {
        synchronized (this.monitor) {
            return this.threadPoolExecutor == null
                    ? this.corePoolSize
                    : this.threadPoolExecutor.getCorePoolSize();
        }
    }

    /**
     * Set the ThreadPoolExecutor's core pool size.
     * Default is 1.
     * <p><b>This setting can be modified at runtime, for example through JMX.</b>
     */
    public void setCorePoolSize(int corePoolSize) {
        synchronized (this.monitor) {
            this.corePoolSize = corePoolSize;
            if (this.threadPoolExecutor != null) {
                this.threadPoolExecutor.setCorePoolSize(corePoolSize);
            }
        }
    }

    /**
     * Return the ThreadPoolExecutor's maximum pool size.
     */
    public int getMaxPoolSize() {
        synchronized (this.monitor) {
            return this.threadPoolExecutor == null
                    ? this.maxPoolSize
                    : this.threadPoolExecutor.getMaximumPoolSize();
        }
    }

    /**
     * Set the ThreadPoolExecutor's maximum pool size.
     * Default is {@code Integer.MAX_VALUE}.
     * <p><b>This setting can be modified at runtime, for example through JMX.</b>
     */
    public void setMaxPoolSize(int maxPoolSize) {
        synchronized (this.monitor) {
            this.maxPoolSize = maxPoolSize;
            if (this.threadPoolExecutor != null) {
                this.threadPoolExecutor.setMaximumPoolSize(maxPoolSize);
            }
        }
    }

    /**
     * Get the capacity for the ThreadPoolExecutor's BlockingQueue.
     * Default is {@code Integer.MAX_VALUE}.
     * <p>Any positive value will lead to a LinkedBlockingQueue instance;
     * any other value will lead to a SynchronousQueue instance.
     *
     * @see java.util.concurrent.LinkedBlockingQueue
     * @see java.util.concurrent.SynchronousQueue
     */
    public int getQueueCapacity() {
        synchronized (this.monitor) {
            return this.threadPoolExecutor == null
                    ? this.queueCapacity
                    : this.threadPoolExecutor.getQueueCapacity();
        }
    }

    /**
     * Set the capacity for the ThreadPoolExecutor's BlockingQueue.
     * Default is {@code Integer.MAX_VALUE}.
     * <p>Any positive value will lead to a LinkedBlockingQueue instance;
     * any other value will lead to a SynchronousQueue instance.
     *
     * @see java.util.concurrent.LinkedBlockingQueue
     * @see java.util.concurrent.SynchronousQueue
     */
    public void setQueueCapacity(int queueCapacity) {
        synchronized (this.monitor) {
            this.queueCapacity = queueCapacity;
            if (this.threadPoolExecutor != null) {
                this.threadPoolExecutor.setQueueCapacity(queueCapacity);
            }
        }
    }

    /**
     * Return the ThreadPoolExecutor's keep-alive seconds.
     */
    public int getKeepAliveSeconds() {
        synchronized (this.monitor) {
            return this.threadPoolExecutor == null
                    ? this.keepAliveSeconds
                    : (int) this.threadPoolExecutor.getKeepAliveTime(TimeUnit.SECONDS);
        }
    }

    /**
     * Set the ThreadPoolExecutor's keep-alive seconds.
     * Default is 60.
     * <p><b>This setting can be modified at runtime, for example through JMX.</b>
     */
    public void setKeepAliveSeconds(int keepAliveSeconds) {
        synchronized (this.monitor) {
            this.keepAliveSeconds = keepAliveSeconds;
            if (this.threadPoolExecutor != null) {
                this.threadPoolExecutor.setKeepAliveTime(keepAliveSeconds, TimeUnit.SECONDS);
            }
        }
    }


    /**
     * Specify whether to allow core threads to time out. This enables dynamic
     * growing and shrinking even in combination with a non-zero queue (since
     * the max pool size will only grow once the queue is full).
     * <p>Default is "false".
     *
     * @see java.util.concurrent.ThreadPoolExecutor#allowCoreThreadTimeOut(boolean)
     */
    public void setAllowCoreThreadTimeOut(boolean allowCoreThreadTimeOut) {
        synchronized (this.monitor) {
            this.allowCoreThreadTimeOut = allowCoreThreadTimeOut;
            if (this.threadPoolExecutor != null) {
                this.threadPoolExecutor.allowCoreThreadTimeOut(allowCoreThreadTimeOut);
            }
        }
    }

    /**
     * Set the RejectedExecutionHandler to use for the ExecutorService.
     * Default is the ExecutorService's default abort policy.
     *
     * @see java.util.concurrent.ThreadPoolExecutor.AbortPolicy
     */
    public void setRejectedExecutionHandler(RejectedExecutionHandler rejectedExecutionHandler) {
        if (rejectedExecutionHandler == null)
            rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();
        synchronized (this.monitor) {
            this.rejectedExecutionHandler = rejectedExecutionHandler;
            if (this.threadPoolExecutor != null) {
                this.threadPoolExecutor.setRejectedExecutionHandler(rejectedExecutionHandler);
            }
        }
    }

    //endregion


    //region executor

    /**
     * Return the underlying ScheduledExecutorService for native access.
     *
     * @return the underlying ScheduledExecutorService (never {@code null})
     * @throws IllegalStateException if the ThreadPoolTaskScheduler hasn't been initialized yet
     */
    @Override
    public ScheduledExecutorService getScheduledExecutor() throws IllegalStateException {
        Assert.state(this.threadPoolExecutor != null, "ThreadPoolTaskScheduler not initialized");
        return this.threadPoolExecutor;
    }

    /**
     * Return the underlying ScheduledThreadPoolExecutor, if available.
     *
     * @return the underlying ScheduledExecutorService (never {@code null})
     * @throws IllegalStateException if the ThreadPoolTaskScheduler hasn't been initialized yet
     *                               or if the underlying ScheduledExecutorService isn't a ScheduledThreadPoolExecutor
     * @see #getScheduledExecutor()
     */
    public ScheduledThreadPoolExecutorEx getScheduledThreadPoolExecutorEx() throws IllegalStateException {
        Assert.state(this.threadPoolExecutor != null, "No ScheduledThreadPoolExecutor available");
        return this.threadPoolExecutor;
    }

    /**
     * @deprecated use getScheduledThreadPoolExecutorEx() instead.
     */
    @Deprecated
    @Override
    public ScheduledThreadPoolExecutor getScheduledThreadPoolExecutor() throws IllegalStateException {
        throw new NotSupportedException();
    }

    //endregion


    //region state

    /**
     * Return the number of currently active threads.
     * <p>Requires an underlying {@link ScheduledThreadPoolExecutor}.
     *
     * @see #getScheduledThreadPoolExecutor()
     * @see ScheduledThreadPoolExecutor#getActiveCount()
     */
    @Override
    public int getActiveCount() {
        return this.threadPoolExecutor == null ? 0 : this.threadPoolExecutor.getActiveCount();
    }

    /**
     * Return the current pool size.
     * <p>Requires an underlying {@link ScheduledThreadPoolExecutor}.
     *
     * @see #getScheduledThreadPoolExecutor()
     * @see ScheduledThreadPoolExecutor#getPoolSize()
     */
    @Override
    public int getPoolSize() {
        return this.threadPoolExecutor == null ? 0 : this.threadPoolExecutor.getPoolSize();
    }

    /**
     * @deprecated not supported.
     */
    @Deprecated
    @Override
    public void setPoolSize(int poolSize) {
        throw new NotSupportedException();
    }

    //endregion
}
